package mapreduce;

import cfar.ff.fun.F1;
import cfar.ff.fun.F2;
import cfar.ff.fun.F3;
import fj.data.List;
import fj.control.parallel.Strategy;
import fj.F; 
import java.util.Timer;
import static fj.data.List.*;
import static fj.control.parallel.Strategy.*;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.lang.Math;
import java.lang.Double;


/**
 *
 * @author haoyun
 */
public class RMap <A,B> implements F3<F1<A,B>, List<A>, List<Integer>, List<B>>{


    public List<B> f(F1<A, B> func, List<A> data, List<Integer> rel) {
        
        int i,id=0;
        List rdata=list();
        for(Integer r : rel){

            for (i=0;i<r;i++){
                rdata=rdata.snoc(data.index(id));
            }
            id++;
        }

        F1 func1=func;
        
        List<B> result = (List<B>) simpleThreadStrategy().parMap(func1, rdata)._1();
        List<B> rresult=list();
        id=0;
        for (Integer r: rel){

            for(i=0;i<r;i++){
                if(result.index(id)!=null){
                    rresult=rresult.snoc(result.index(id));
                    id=id+r-i;
                    break;
                }
                else if(result.index(id)==null&i==r-1)
                    rresult=rresult.snoc(result.index(id));
                id++;
            }
        }
        return rresult;

    }

    public List<B> f(F1<A,B> func, List<A> data, List<Double> rel, List<Double> err){

        List <Integer> num=Reliable.prob2num(rel,err);
        return this.f(func,data,num);


    }

    public List<B> f(F1<A, B> func, List<A> data){

        F1 func1=func;
        
        return (List<B>) simpleThreadStrategy().parMap(func1, data)._1();
    }

    public int prob2num(double rel, double err){

        int num=(int)Math.round(Math.log(1-rel)/Math.log(err));
        System.out.println(num+" replications of this job are needed");
        return num;
    }

    public static void main(String[] args){
//        double rel=0.9;
//        double err=0.3;
//        RMap map=new RMap();
//        map.prob2num(rel,err);
        F1 ff = new F1<Integer, Integer>() {

        public Integer f(Integer a) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    Logger.getLogger(RMap.class.getName()).log(Level.SEVERE, null, ex);
                }
                return a + 10;
            }
        };

        List l=list(1,2);

        List<Double> rel=list(0.9,0.9);
        List<Double> err=list(0.3,0.1);
        RMap map=new RMap();
        List result=map.f(ff, l, rel,err);
        int id=0;
        for (Object r: result){
            System.out.println(r);
        }
    }

}
